package data.platform.stream.internal.command;

import data.platform.common.domain.MetricValue;
import data.platform.common.service.query.MetricTagQueryService;
import data.platform.common.stream.aggregates.Rule;
import data.platform.common.stream.aggregates.TimeWindow;
import data.platform.common.stream.aggregates.WindowAggregateItem;
import data.platform.common.util.DateUtil;
import data.platform.stream.internal.cache.RuleAggregateCache;
import data.platform.stream.internal.cache.RuleCache;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;

@Slf4j
@Service
@AllArgsConstructor
public class TimeWindowedService {

    final RuleCache ruleCache;

    final MetricTagQueryService metricTagQueryService;

    final RuleAggregateCache ruleAggregateCache;

    public Mono<Integer> process(MetricValue metricValue) {
        try {
            List<Rule> rules = ruleCache.filterRules(metricValue);
            return Flux.fromIterable(rules)
                    .flatMap(rule -> timeWindowAggregate(rule, metricValue))
                    .collectList()
                    .map(values -> values.size());
        } catch (Exception ex) {
            log.error("", ex);
            return Mono.empty();
        }


    }

    private Mono<Integer> timeWindowAggregate(Rule rule, MetricValue metricValue) {
        Optional<Integer> tagOptional = metricTagQueryService.findTagId(metricValue.getTagJson());
        Mono<Integer> tagMono;
        if (tagOptional.isPresent()) {
            tagMono = Mono.just(tagOptional.get());
        } else {
            tagMono = Mono.defer(() -> Mono.delay(Duration.ofMillis(3000))
                    .map(i -> metricTagQueryService.findTagId(metricValue.getTagJson()).get()));
        }
        return tagMono.zipWhen(tagId -> Mono.just(metricValue)
                .doOnNext(mv -> {
                    if (ruleAggregateCache.getCache().containsKey(rule.getId())) {
                        long eventTime = DateUtil.getDateOfLocalDateTime(metricValue.getEventTime()).getTime();

                        Map<TimeWindow, Map<String, Map<Integer, WindowAggregateItem>>> timeWindowMap = ruleAggregateCache.getCache().get(rule.getId());
                        if (!timeWindowMap.isEmpty()) {
                            Set<TimeWindow> timeWindowSet = timeWindowMap.keySet();
                            for (TimeWindow timeWindow : timeWindowSet) {
                                if (timeWindow.getStart() <= eventTime && timeWindow.getEnd() > eventTime) {
                                    compute(rule.getId(), tagId, timeWindow, metricValue);
                                    return;
                                }
                            }

                            // 获取当前任务的时间窗口，如果时间窗口存在且事件时间早于最小开始时间，数据已过期，不做处理
                            long windowEarlyStartTime = Long.MAX_VALUE;
                            for (TimeWindow timeWindow : timeWindowSet) {
                                if (timeWindow.getStart() < windowEarlyStartTime) {
                                    windowEarlyStartTime = timeWindow.getStart();
                                }
                            }
                            if (eventTime < windowEarlyStartTime) {
                                return;
                            }
                        }

                        TimeWindow timeWindow = TimeWindow.getTimeWindow(rule.getWindowType(), rule.getWindowSize(), metricValue.getEventTime());
                        compute(rule.getId(), tagId, timeWindow, metricValue);
                    } else {
                        TimeWindow timeWindow = TimeWindow.getTimeWindow(rule.getWindowType(), rule.getWindowSize(), metricValue.getEventTime());
                        compute(rule.getId(), tagId, timeWindow, metricValue);
                    }
                }))
                .map(tuple2 -> 1);
    }

    private void compute(String ruleId, Integer tagId, TimeWindow timeWindow, MetricValue metricValue) {
        Map<Integer, WindowAggregateItem> tagIdCache = ruleAggregateCache.getTagIdTimeWindowCache(ruleId, timeWindow, metricValue.getMetric());
        WindowAggregateItem aggregateItem = tagIdCache.computeIfAbsent(tagId, k -> new WindowAggregateItem(Double.valueOf(0.0), Double.NEGATIVE_INFINITY, Double.POSITIVE_INFINITY, Double.valueOf(0.0), 0));
        aggregateItem.compute(metricValue.getValue(), metricValue.getEventTime());
    }
}
